package org.chnxi.study.messagequeue.rockermq.producer;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.ToString;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.ArrayList;
import java.util.List;

/**
 * 在符合规则下，将同一类的消息放入到同一个queue中，在一个queue中使用顺序发送，这样可以保证一个queue中的消息顺序性
 */
public class OrderMsgProducer {

    static List<BusiData> datas = new ArrayList()
    {{
        add(new BusiData(1 , "message1-1"));
        add(new BusiData(2 , "message2-1"));
        add(new BusiData(1 , "message1-2"));
        add(new BusiData(2 , "message2-2"));
        add(new BusiData(1 , "message1-3"));
        add(new BusiData(2 , "message2-3"));
        add(new BusiData(1 , "message1-4"));
        add(new BusiData(2 , "message2-4"));
    }};


    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //使用默认提供的Producer
        DefaultMQProducer producer = new DefaultMQProducer();
        //设置NameServer地址
        producer.setNamesrvAddr("192.168.50.227:9876;192.168.50.227:9877");
        //启动Producer
        producer.start();

        for (BusiData busiData : datas) {
            Message msg = new Message("topicName" , "tag" , busiData.toString().getBytes());
            //发送消息时指定MessageQueue的选择器，保证将同一业务的消息发送到一个MessageQueue中
            producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    long busiId = (long)o;
                    return list.get((int)(busiId%list.size()));  //使用业务ID对list取余的方式
                }
            }, busiData.busiId);
        }

        producer.shutdown();

    }

    @AllArgsConstructor
    @ToString
    static class BusiData{
        long busiId;
        String busiMsg;
    }

}
